#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#ifdef HAVE_EPOLL
#include <sys/epoll.h>
#endif

#include <unistd.h>
#include <fcntl.h>
#include <errno.h>

#include "log.h"
#include "s2a.h"

static int setnblock(int socket);

typedef struct Message
{
    uint64_t sn;
    time_t timeout;
    const char * request;
    int request_len;
    int result;
    char * respond;
    int respond_len;
} Message;

#define MAX_WORKER_CNT 1024
#define MAX_THREAD_CNT 32
#define MAX_QUEUE_SIZE 1024

typedef struct MessageQueue {
    Message * msg;
    int size;
    int head;
    int tail;
} MessageQueue;

static bool is_queue_full(MessageQueue * queue)
{
    return (((queue->tail+1)%queue->size) == queue->head);
}

static bool is_queue_empty(MessageQueue * queue)
{
    return (queue->tail == queue->head);
}

static bool queue_push_message(MessageQueue * queue, Message * msg)
{
    if (is_queue_full(queue)) {
        return false;
    }

    Message * m = queue->msg + queue->tail;
    m->sn = msg->sn;
    m->timeout = msg->timeout;
    m->request = msg->request;
    m->request_len = msg->request_len;
    m->result = msg->result;
    m->respond = msg->respond;
    m->respond_len = msg->respond_len;

    queue->tail = (queue->tail+1)%queue->size;
    return true;
}

static bool queue_pop_message(MessageQueue * queue, Message * msg)
{
    if (is_queue_empty(queue)) {
        return false;
    }

    Message * m = queue->msg + queue->head;
    msg->sn = m->sn;
    msg->timeout = m->timeout;
    msg->request = m->request;
    msg->request_len = m->request_len;
    msg->result = m->result;
    msg->respond = m->respond;
    msg->respond_len = m->respond_len;

    queue->head = (queue->head+1)%queue->size;
    return true;
}

static void queue_free(MessageQueue * queue) {
    if (queue->msg) free(queue->msg);
    queue->msg = 0;
    queue->size = 0;
    queue->head = queue->tail = 0;
}

static int queue_init(MessageQueue * queue, int size)
{
    queue_free(queue);
    if (queue->msg) free(queue->msg);
    queue->msg = malloc(sizeof(Message)*size);
    if (queue->msg == 0) {
        return -1;
    }
    queue->size = size;
    return 0;
}

typedef struct WorkerDetail
{
    int read_fd;
    int write_fd;

    s2a_work_func work;
    void * data;

    MessageQueue request_queue;
    pthread_mutex_t  mutex;
    pthread_cond_t   cond;

    MessageQueue respond_queue;

    pthread_t threads[MAX_THREAD_CNT];
}WorkerDetail;


static WorkerDetail g_workers[MAX_WORKER_CNT];

bool worker_init()
{
    int i;
    for(i = 0; i < MAX_WORKER_CNT; i++) {
        WorkerDetail * wd = g_workers+i;
        wd->read_fd = wd->write_fd = -1;
        wd->work = 0;
        wd->data = 0;
        memset(wd->threads, 0, sizeof(pthread_t) * MAX_THREAD_CNT);

        wd->request_queue.msg = 0;
        wd->request_queue.size = 0;
        wd->request_queue.head = 0;
        wd->request_queue.tail = 0;

        wd->request_queue.msg = 0;
        wd->request_queue.size = 0;
        wd->request_queue.head = 0;
        wd->request_queue.tail = 0;
    }

    return true;
}

static const char * g_exit_flag = "work_exit";

static void * work_thread(void * p)
{
    WorkerDetail * worker = (WorkerDetail*)p;

    while(1) {
        Message msg;
        pthread_mutex_lock(&worker->mutex);
        if (!queue_pop_message(&worker->request_queue, &msg)) {
            pthread_cond_wait(&worker->cond, &worker->mutex);
            pthread_mutex_unlock(&worker->mutex);
            continue;
        }
        pthread_mutex_unlock(&worker->mutex);

        if (msg.request == g_exit_flag) {
            break;
        }

        if (msg.timeout < time(0)) {
            msg.result = -1;
        } else {
            msg.result = worker->work(msg.request, msg.request_len,
                    &msg.respond, &msg.respond_len, worker->data);
        }

        pthread_mutex_lock(&worker->mutex);
        while (!queue_push_message(&worker->respond_queue, &msg)) {
            pthread_mutex_unlock(&worker->mutex);
            usleep(100);
            pthread_mutex_lock(&worker->mutex);
        }
        pthread_mutex_unlock(&worker->mutex);
        int n = write(worker->write_fd, " ", 1);
        WRITE_DEBUG_LOG("write %d byte to %d\n", n, worker->write_fd);
    }
    WRITE_DEBUG_LOG("worker_thread exit\n");
    return 0;
}

static int setnblock(int socket)
{
	int flags;
	flags = fcntl(socket, F_GETFL, 0);
	if (flags == -1) {
		return -1;
	}

	flags |= O_NONBLOCK;
	flags |= O_NDELAY;
	if (fcntl(socket, F_SETFL, flags) == -1) {
		return -1;
	}
	return 0;
}

Worker * worker_create(s2a_work_func work, int thread_cnt, void * data)
{
    if (work == 0 || thread_cnt <= 0 || thread_cnt >= MAX_THREAD_CNT) {
        return 0;
    }

    int i;
    WorkerDetail * worker = 0;
    for (i = 0; i < MAX_WORKER_CNT; i++) {
        if (g_workers[i].work == 0
                && g_workers[i].read_fd == -1
                && g_workers[i].write_fd == -1) {
            worker = g_workers+i;
            break;
        }
    }

    int mutex_init = 0;
    int cond_init = 0;

    if (worker) {
        int pfd[2] = {-1, -1};
        if (pipe(pfd) == -1) {
            return 0;
        }

        setnblock(pfd[1]);

        worker->read_fd = pfd[0];
        worker->write_fd = pfd[1];

        worker->work = work;

        if (queue_init(&worker->request_queue, MAX_QUEUE_SIZE) != 0) {
            goto clean;
        }

        if (queue_init(&worker->respond_queue, MAX_QUEUE_SIZE) != 0) {
            goto clean;
        }

        if (pthread_mutex_init(&worker->mutex, NULL) != 0) {
            goto clean;
        }
        mutex_init = 1;

        if (pthread_cond_init(&worker->cond, NULL) != 0) {
            goto clean;
        }
        cond_init = 1;

        for (i = 0; i < thread_cnt; i++) {
            if (pthread_create(&(worker->threads[i]),
                        0, work_thread, worker) != 0) {
                worker->threads[i] = 0;
            }
        }
    }
    return (Worker*)worker;
clean :

    if (worker) {
        for(i = 0; i < thread_cnt; i++ ) {
            if (worker->threads[i]) {
                pthread_cancel(worker->threads[i]);
            }
        }

        if (mutex_init != 0) {
            pthread_mutex_destroy(&worker->mutex);
        }

        if (cond_init != 0) {
            pthread_cond_destroy(&worker->cond);
        }

        queue_free(&worker->request_queue);
        queue_free(&worker->respond_queue);

        if (worker->read_fd >= 0)  close(worker->read_fd);
        if (worker->write_fd >= 0) close(worker->write_fd);
        worker->read_fd = worker->write_fd = -1;
        worker->work = 0;
    }
    return 0;
}

bool post_request(Worker * worker,
                  uint64_t sn,
                  int timeout_sec,
                  const char * request,
                  int request_len) {
    WorkerDetail * wd = (WorkerDetail*)worker;
    if (wd < g_workers || wd >= g_workers+MAX_WORKER_CNT) {
        return false;
    }

    Message msg;
    msg.sn = sn;
    msg.timeout = timeout_sec + time(0);
    msg.request = request;
    msg.request_len = request_len;
    msg.result = 0;
    msg.respond = 0;
    msg.respond_len = 0;

    pthread_mutex_lock(&wd->mutex);
    bool ret = queue_push_message(&wd->request_queue, &msg);
    pthread_cond_signal(&wd->cond);
    pthread_mutex_unlock(&wd->mutex);
    return ret;
}

bool get_respond(Worker * worker,
                 uint64_t * sn,
                 int * result,
                 const char ** request,
                 int * request_len,
                 char ** respond,
                 int * respond_len)
{
    WorkerDetail * wd = (WorkerDetail*)worker;
    if (wd < g_workers || wd >= g_workers+MAX_WORKER_CNT) {
        return false;
    }

    if (wd->work == 0) {
        return false;
    }

    char buff[1];
    read(wd->read_fd, buff, 1);

    Message msg;

    pthread_mutex_lock(&wd->mutex);
    int ret = queue_pop_message(&wd->respond_queue, &msg);
    pthread_mutex_unlock(&wd->mutex);
    if (ret){
        if(sn) *sn = msg.sn;
        if(result) *result = msg.result;
        if(request) *request = msg.request;
        if(request_len) *request_len = msg.request_len;
        if(respond) *respond = msg.respond;
        if(respond_len) *respond_len = msg.respond_len;
    }
    return ret;
}

int worker_destory(Worker * worker)
{
    WorkerDetail * wd = (WorkerDetail*)worker;
    if (wd < g_workers || wd >= g_workers+MAX_WORKER_CNT) {
        return -1;
    }

    int i;
    for (i = 0; i < MAX_THREAD_CNT; i++) {
        while(!post_request(worker, 0, 100, g_exit_flag, strlen(g_exit_flag))) {
            usleep(100);
        }
    }

    for(i = 0;i < MAX_THREAD_CNT; i++) {
        if(wd->threads[i] != 0 ) {
            void * p = 0;
            pthread_join(wd->threads[i], &p);
        }
    }

    wd->work = 0;
    pthread_mutex_destroy(&wd->mutex);
    pthread_cond_destroy(&wd->cond);

    queue_free(&wd->request_queue);
    queue_free(&wd->respond_queue);

    if (wd->read_fd >= 0)  close(wd->read_fd);
    if (wd->write_fd >= 0) close(wd->write_fd);
    wd->data = 0;
    wd->read_fd = wd->write_fd= -1;
    return 0;
}
